package cn.itcast.up.model.statistisc

import cn.itcast.up.model.base.BaseModel
import cn.itcast.up.model.statistisc.AgeModel.{computeTagIds, getFiveDF, getFourRule, getMysqlSource, loadHBaseData, loadOldData, mergeTotalTag, sinkData, spark}
import org.apache.spark.sql.{Column, DataFrame}

/**
  * 用户消费周期指标计算
  */
object CycleModel extends BaseModel{

  import spark.implicits._
  import org.apache.spark.sql.functions._
  /**
    * 设置当前应用的名字
    *
    * @return
    */
  override def setAppName(): String = "CycleModel"

  /**
    * 设置4级标签ID
    *
    * @return
    */
  override def setFourTagID(): Int = 499

  /**
    * 标签计算
    *
    * @param hbaseDF HBase数据源
    * @param fiveDF  规则数据
    * @return 最终的计算结果:userid, tagIds
    */
  override def computeTagIds(hbaseDF: DataFrame, fiveDF:  DataFrame): DataFrame = {
//    fiveDF.show()
//    +---+-------+
    //| id|   rule|
    //+---+-------+
    //|500|    0-7|
    //|501|   8-14|
    //|502|  15-30|
    //|503|  31-60|
    //|504|  61-90|
    //|505| 91-120|
    //|506|121-150|
    //|507|151-180|
    //+---+-------+
    //处理5级规则数据
    val ruleDF: DataFrame = fiveDF.map(row => {
      val tagIds: String = row.getAs[Long]("id").toString
      val arr: Array[String] = row.getAs[String]("rule").split("-")
      (tagIds, arr(0), arr(1))
    }).collect().toList.toDF("tagIds", "start", "end")

//    hbaseDF.show()
//+---------+----------+
    //| memberId|finishTime|
    //+---------+----------+
    //| 13823431|1564415022|
    //| 13823431|1564425022|
    //|  4035167|1565687310|
    //|  4035291|1564681801|
    //|  4035041|1565799378|


    //按照用户ID键分组
    val groupDF: DataFrame = hbaseDF.groupBy('memberId)
      //获取用户订单完成的最大值.最后一笔订单
      .agg(max('finishTime).as("finishTime"))

    val dayNum: Column = datediff(
      //      当前时间-订单完成时间
      current_timestamp(),
      //将订单完成的时间戳转换为日期
      from_unixtime(groupDF.col("finishTime"))
    )
//      .show()
//+---------+---------------+
    //| memberId|     finishTime|
    //+---------+---------------+
    //| 13822725|     1566056954|
    //| 13823083|     1566048648|
    //|138230919|     1566012606|
    //| 13823681|     1566012541|
      val dayNumDF: DataFrame = groupDF.select('memberId.as("userid"), dayNum.as("dayNum"))
//      .show()
//+---------+------+
    //|   userid|dayNum|
    //+---------+------+
    //| 13822725|   122|
    //| 13823083|   122|
    //|138230919|   122|
    //| 13823681|   122|
    //|  4033473|   122|
    dayNumDF.join(ruleDF)
      .where('dayNum.between('start,'end))
//      .show()
//+---------+------+------+-----+---+
    //|   userid|dayNum|tagIds|start|end|
    //+---------+------+------+-----+---+
    //| 13822725|   122|   506|  121|150|
    //| 13823083|   122|   506|  121|150|
    //|138230919|   122|   506|  121|150|
    //| 13823681|   122|   506|  121|150|
      .select('userid, 'tagIds)
  }


  def main(args: Array[String]): Unit = {
    //加载MySQL数据源
    val mysqlSource: DataFrame = getMysqlSource
    //获取4级规则的map
    val map: Map[String, String] = getFourRule(mysqlSource)
    //获取5级规则的DataFrame
    val fiveDF: DataFrame = getFiveDF(mysqlSource)
    //加载HBase数据源
    val hbaseSource: DataFrame = loadHBaseData(map)
    //开始进行标签计算
    val newDF: DataFrame = computeTagIds(hbaseSource, fiveDF)
    //加载历史数据
    val oldDF: DataFrame = loadOldData(map)
    //进行标签合并
    val result: DataFrame = mergeTotalTag(newDF, oldDF)
    //将合并之后的数据落地
    sinkData(result, map)
  }
}
